package cn._51doit.flink.day06;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;

public class ActivityCountFunction extends KeyedProcessFunction<Tuple2<String, String>, Tuple3<String, String, String>, Tuple4<String, String, Integer, Integer>> {

    //transient代表瞬时的，用该关键字修饰的字段不参与序列化的反序列化
    //但是在checkpoint时，会将State中的数据保存到状态存储后端
    private transient ValueState<Integer> countState;
    private transient ValueState<HashSet<String>> distinctCountState;
    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化或恢复状态
        ValueStateDescriptor<Integer> countStateDesc = new ValueStateDescriptor<Integer>("count-state", Integer.class);
        countState = getRuntimeContext().getState(countStateDesc);

        ValueStateDescriptor<HashSet<String>> distinctCountStateDesc = new ValueStateDescriptor<>("dis-count-state", TypeInformation.of(new TypeHint<HashSet<String>>() {}));
        distinctCountState = getRuntimeContext().getState(distinctCountStateDesc);
    }

    @Override
    public void processElement(Tuple3<String, String, String> value, Context ctx, Collector<Tuple4<String, String, Integer, Integer>> out) throws Exception {

        //统计次数
        Integer count = countState.value();
        if (count == null) {
            count = 0;
        }
        count += 1;
        countState.update(count);
        //统计人数
        HashSet<String> uids = distinctCountState.value();
        if (uids == null) {
            uids = new HashSet<>();
        }
        uids.add(value.f0); //set可以去重
        //更新
        distinctCountState.update(uids);

        //输出数据
        out.collect(Tuple4.of(value.f1, value.f2, count, uids.size()));
    }
}
